package cdrprocessing.bolt.rule;

import java.io.PrintWriter;
import java.net.Socket;
import java.util.HashMap;
import java.util.Map;

import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseRichBolt;
import backtype.storm.tuple.Tuple;

import com.fasterxml.jackson.databind.ObjectMapper;

public class ExportResultBolt extends BaseRichBolt {

    private PrintWriter output;

    private ObjectMapper jsonParser;

    public void declareOutputFields(OutputFieldsDeclarer declarer) {
    }

    public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
        try {
            output = new PrintWriter(new Socket("localhost", 8181).getOutputStream());
            output.println("ruleResult");
            output.flush();

            jsonParser = new ObjectMapper();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

    public void execute(Tuple input) {
        final Map<String, Object> data = new HashMap<String, Object>();

        data.put("rule", input.getValueByField("rule"));

        if ("DiscountForPopularUsersRule".equals(input.getSourceStreamId())) {
            data.put("phoneNumber", input.getValueByField("phoneNumber"));
        } else if ("DiscountForManyCallsRule".equals(input.getSourceStreamId())) {
            data.put("phoneNumber", input.getValueByField("phoneNumber"));
        } else if ("DiscountForFailureRule".equals(input.getSourceStreamId())) {
            data.put("areaCode", input.getValueByField("areaCode"));
        } else {
            System.out.println("Unexpected tuple " + input);
            return;
        }

        try {
            output.println(jsonParser.writeValueAsString(data));
            output.flush();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

}
